package edu.npu.GraphRedundance;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import edu.npu.GraphTransform.GraphSlim;

public class QuasiHopRedundanceDriver {
	final static String SEPARATOR = "\t";
	public static final String USERNAME = "zhao";
	public static final String PATH_PREFIX = "hdfs://test118:9000/user/" + USERNAME + "/";
	
	public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException {
		Configuration conf = new Configuration();
		Job job = new Job(conf, "quasiHop");
		job.setJarByClass(TriHopRedundanceDriver.class);
		job.setMapperClass(TriHopRedundanceDriver.TriHopRedundanceMapper.class);
		job.setReducerClass(TriHopRedundanceDriver.TriHopRedundanceReducer.class);
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(Text.class);
		job.setNumReduceTasks(12);
		FileInputFormat.addInputPath(job, new Path(PATH_PREFIX + "graph_part"));
		FileInputFormat.addInputPath(job, new Path(PATH_PREFIX + "triHop"));
		FileOutputFormat.setOutputPath(job, new Path(PATH_PREFIX + "quasiHop"));
		job.waitForCompletion(true);
		
		Job slim = new Job(conf, "triSlim");
		slim.setJarByClass(GraphSlim.class);
		slim.setMapperClass(GraphSlim.GraphSlimMapper.class);
		slim.setReducerClass(GraphSlim.GraphSlimReducer.class);
		slim.setOutputKeyClass(Text.class);
		slim.setOutputValueClass(Text.class);
		slim.setNumReduceTasks(12);
		FileInputFormat.addInputPath(slim, new Path(PATH_PREFIX + "triHop_slim"));
		FileInputFormat.addInputPath(slim, new Path(PATH_PREFIX + "quasiHop"));
		FileOutputFormat.setOutputPath(slim, new Path(PATH_PREFIX + "quasiHop_slim"));
		slim.waitForCompletion(true);
	}
}
